package cn.edu.nchu.service.impl;

import cn.edu.nchu.service.MQService;
import cn.edu.nchu.service.MqttService;
import com.alibaba.fastjson.JSONObject;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

@Service
public class MqttServiceImpl implements MqttService {
    @Autowired
    private MQService mqService;

    private static String broker = "ws://175.24.184.18:8083/mqtt";
    private static final String username = "test";
    private static String password= "test";
    private MqttClient client;
    private final String TOPIC_PREFIX = "/+/+/";
    private static final String[] SUSCRIBE_TOPICS = {"messages/up", "commands/response"};


    @Override
    @PostConstruct // 应用启动时调用该方法
    @Async // 异步线程
    public void startMqttListner() throws MqttException {
        initClient(mqService);
        connectMqtt();
        // 订阅
        for (String topic : SUSCRIBE_TOPICS) {
            client.subscribe(TOPIC_PREFIX + topic);
        }
    }

    private void connectMqtt() throws MqttException {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        connectOptions.setUserName(username);
        connectOptions.setPassword(password.toCharArray());
        // 保留会话
        connectOptions.setCleanSession(true);
        // 建立连接
        client.connect(connectOptions);
    }

    private void initClient(MQService mqService) throws MqttException {
        client = new MqttClient(broker, username, new MemoryPersistence());
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                // 连接丢失后，一般在这里面进行重连
                try {
                    client.reconnect();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }

            // 确保消息是json格式不然会出错
            @Override
            public void messageArrived(String topic, MqttMessage message) {
                // subscribe后得到的消息会执行到这里面
                // 得到消息类型
                String[] s = topic.split("/");
                String message_type = s[3]+"/"+s[4];
                // 给消息队列发消息
                mqService.sendMessage(message_type, JSONObject.parseObject(new String(message.getPayload())));
                System.out.println(topic);

                System.out.println(message);
//                mqService.doNothing(JSONObject.parseObject(new String(message.getPayload())));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                // 发布消息后要做的事
            }
        });
    }

    @Override
    public void sendMessage(String topic, String content) {
        try {
            MqttMessage message = new MqttMessage(content.getBytes());
            client.publish(topic, message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
